package online.shenjian.hadoop.reducejoin;

import org.apache.avro.Schema;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author shenjian
 * @since 2023/2/25
 */
public class OrderProductReducer extends Reducer<Text, OrderProductDto, OrderProductDto, NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<OrderProductDto> values, Reducer<Text, OrderProductDto, OrderProductDto, NullWritable>.Context context) throws IOException, InterruptedException {
        // 存放所有订单信息
        List<OrderProductDto> orderDtoList = new ArrayList<>();
        // 存放产品信息
        OrderProductDto productDto = new OrderProductDto();

        for (OrderProductDto value : values) {
            if ("order".equals(value.getFlag())) {
                // 防止因为指向同一内存地址导致覆盖
                OrderProductDto temp = new OrderProductDto();
                try {
                    BeanUtils.copyProperties(temp, value);
                } catch (IllegalAccessException | InvocationTargetException e) {}
                orderDtoList.add(temp);
            } else if ("product".equals(value.getFlag())) {
                try {
                    BeanUtils.copyProperties(productDto, value);
                } catch (IllegalAccessException | InvocationTargetException e) {}
            }
        }
        // 循环遍历赋值订单表产品名称
        for (OrderProductDto dto : orderDtoList) {
            dto.setProductName(productDto.getProductName());
            context.write(dto, NullWritable.get());
        }
    }
}
